package juc.scenarios.pipeline;

import lombok.Getter;
import lombok.Setter;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 图片处理管道演示 - 综合使用多种并发工具
 * <p>
 * 功能特点：
 * 1. 图片下载：Semaphore控制并发下载数量
 * 2. 缓存管理：ReadWriteLock实现读写缓存控制
 * 3. 任务队列：BlockingQueue实现生产者-消费者模式
 * 4. 分块处理：Fork/Join框架进行图片分块压缩
 * 5. 阶段同步：CyclicBarrier实现多阶段处理同步
 * 6. 状态监控：AtomicReference实现状态原子更新
 * 7. 性能统计：AtomicLong实现线程安全统计
 */
public class PicPipelineDemo {

    // 基础配置
    private static final int DOWNLOAD_THREADS = 3;      // 下载线程数
    private static final int PROCESS_THREADS = 4;         // 处理线程数
    private static final int TOTAL_TASKS = 12;          // 总任务数
    private static final int CHUNK_SIZE = 1024 * 100;     // 分块大小 100KB

    // 并发控制工具
    private static final Semaphore DOWNLOAD_SEMAPHORE = new Semaphore(DOWNLOAD_THREADS);
    private static final CountDownLatch DOWNLOAD_LATCH = new CountDownLatch(TOTAL_TASKS);
    private static final CyclicBarrier PROCESS_BARRIER = new ProcessBarrier(PROCESS_THREADS);

    // 数据结构和缓存
    private static final BlockingQueue<PicTask> downloadQueue = new LinkedBlockingQueue<>();
    private static final BlockingQueue<PicTask> processQueue = new LinkedBlockingQueue<>();
    private static final Map<String, PicMetadata> cache = new ConcurrentHashMap<>();
    private static final ReadWriteLock cacheLock = new ReentrantReadWriteLock();

    // 统计和监控
    private static final AtomicLong totalDownloadSize = new AtomicLong(0);
    private static final AtomicLong totalProcessTime = new AtomicLong(0);
    private static final AtomicReference<PipelineStatus> status = new AtomicReference<>(PipelineStatus.INIT);

    // 线程池
    private static final ExecutorService downloadPool = Executors.newFixedThreadPool(DOWNLOAD_THREADS);
    private static final ExecutorService processPool = Executors.newFixedThreadPool(PROCESS_THREADS);
    private static final ForkJoinPool forkJoinPool = new ForkJoinPool();

    // 图片任务类
    @Setter
    @Getter
    static class PicTask {
        // Getters and setters
        private final String id;
        private final String url;
        private byte[] data;
        private long downloadTime;
        private long processTime;
        private PicStatus status = PicStatus.PENDING;

        public PicTask(String id, String url) {
            this.id = id;
            this.url = url;
        }

    }

    // 图片元数据类
    @Getter
    static class PicMetadata {
        private final String id;
        private final long size;
        private final long timestamp;
        private final String format;

        public PicMetadata(String id, long size, String format) {
            this.id = id;
            this.size = size;
            this.timestamp = System.currentTimeMillis();
            this.format = format;
        }

        public boolean isExpired() {
            return System.currentTimeMillis() - timestamp > 60000; // 1分钟过期
        }
    }

    // 状态枚举
    enum PipelineStatus {
        INIT, DOWNLOADING, PROCESSING, COMPRESSING, COMPLETED, ERROR
    }

    enum PicStatus {
        PENDING, DOWNLOADING, DOWNLOADED, PROCESSING, PROCESSED, COMPRESSED, ERROR
    }

    // 自定义CyclicBarrier
    static class ProcessBarrier extends CyclicBarrier {
        public ProcessBarrier(int parties) {
            super(parties, () -> {
                System.out.println("🔄 阶段处理完成，准备进入下一阶段...");
                // 可以在这里添加阶段转换逻辑
            });
        }
    }

    // 图片下载器
    static class PicDownloader implements Runnable {
        @Override
        public void run() {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    PicTask task = downloadQueue.poll(1, TimeUnit.SECONDS);
                    if (task == null) continue;

                    downloadPicture(task);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("下载线程被中断");
            }
        }

        private void downloadPicture(PicTask task) {
            DOWNLOAD_SEMAPHORE.acquireUninterruptibly();
            try {
                task.setStatus(PicStatus.DOWNLOADING);
                long startTime = System.currentTimeMillis();

                // 模拟下载过程
                Thread.sleep(ThreadLocalRandom.current().nextInt(200, 800));

                // 生成模拟图片数据
                int size = ThreadLocalRandom.current().nextInt(50000, 500000); // 50KB-500KB
                byte[] data = new byte[size];
                ThreadLocalRandom.current().nextBytes(data);

                task.setData(data);
                task.setDownloadTime(System.currentTimeMillis() - startTime);
                task.setStatus(PicStatus.DOWNLOADED);

                // 更新统计
                totalDownloadSize.addAndGet(size);

                // 添加到处理队列
                processQueue.put(task);

                // 缓存元数据
                cacheMetadata(task.getId(), size, "JPEG");

                System.out.printf("✅ 下载完成: %s, 大小: %.2f KB, 耗时: %d ms%n",
                        task.getId(), size / 1024.0, task.getDownloadTime());

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                task.setStatus(PicStatus.ERROR);
            } finally {
                DOWNLOAD_SEMAPHORE.release();
                DOWNLOAD_LATCH.countDown();
            }
        }

        private void cacheMetadata(String id, long size, String format) {
            cacheLock.writeLock().lock();
            try {
                cache.put(id, new PicMetadata(id, size, format));
            } finally {
                cacheLock.writeLock().unlock();
            }
        }
    }

    // 图片处理器
    static class PicProcessor implements Runnable {
        @Override
        public void run() {
            try {
                PROCESS_BARRIER.await(); // 等待所有处理器就绪

                while (!Thread.currentThread().isInterrupted()) {
                    PicTask task = processQueue.poll(1, TimeUnit.SECONDS);
                    if (task == null) {
                        // 检查是否所有下载任务都已完成
                        if (DOWNLOAD_LATCH.getCount() == 0 && processQueue.isEmpty()) {
                            break;
                        }
                        continue;
                    }

                    processPicture(task);
                }

                PROCESS_BARRIER.await(); // 等待处理阶段完成

            } catch (Exception e) {
                System.err.println("处理器异常: " + e.getMessage());
            }
        }

        private void processPicture(PicTask task) {
            try {
                task.setStatus(PicStatus.PROCESSING);
                long startTime = System.currentTimeMillis();

                // 检查缓存
                if (checkCache(task.getId())) {
                    System.out.printf("💰 缓存命中: %s%n", task.getId());
                    task.setStatus(PicStatus.PROCESSED);
                    return;
                }

                // 使用Fork/Join进行分块处理
                byte[] processedData = processWithForkJoin(task.getData());

                task.setData(processedData);
                task.setProcessTime(System.currentTimeMillis() - startTime);
                task.setStatus(PicStatus.PROCESSED);

                // 更新统计
                totalProcessTime.addAndGet(task.getProcessTime());

                System.out.printf("🔧 处理完成: %s, 压缩率: %.1f%%, 耗时: %d ms%n",
                        task.getId(),
                        (1 - processedData.length / (double) task.getData().length) * 100,
                        task.getProcessTime());

            } catch (Exception e) {
                System.err.println("处理失败: " + task.getId() + " - " + e.getMessage());
                task.setStatus(PicStatus.ERROR);
            }
        }

        private boolean checkCache(String id) {
            cacheLock.readLock().lock();
            try {
                PicMetadata metadata = cache.get(id);
                return metadata != null && !metadata.isExpired();
            } finally {
                cacheLock.readLock().unlock();
            }
        }

        private byte[] processWithForkJoin(byte[] data) {
            ImageCompressionTask task = new ImageCompressionTask(data, 0, data.length, CHUNK_SIZE);
            return forkJoinPool.invoke(task);
        }
    }

    // Fork/Join图片压缩任务
    static class ImageCompressionTask extends RecursiveTask<byte[]> {
        private final byte[] data;
        private final int start;
        private final int end;
        private final int threshold;

        public ImageCompressionTask(byte[] data, int start, int end, int threshold) {
            this.data = data;
            this.start = start;
            this.end = end;
            this.threshold = threshold;
        }

        @Override
        protected byte[] compute() {
            if (end - start <= threshold) {
                // 直接处理小块数据
                return compressChunk(data, start, end);
            } else {
                // 分割任务
                int mid = start + (end - start) / 2;
                ImageCompressionTask leftTask = new ImageCompressionTask(data, start, mid, threshold);
                ImageCompressionTask rightTask = new ImageCompressionTask(data, mid, end, threshold);

                leftTask.fork();
                byte[] rightResult = rightTask.compute();
                byte[] leftResult = leftTask.join();

                return mergeResults(leftResult, rightResult);
            }
        }

        private byte[] compressChunk(byte[] data, int start, int end) {
            // 模拟压缩算法
            byte[] compressed = new byte[(end - start) / 2];
            for (int i = 0; i < compressed.length; i++) {
                compressed[i] = (byte) (data[start + i * 2] >> 1);
            }
            return compressed;
        }

        private byte[] mergeResults(byte[] left, byte[] right) {
            byte[] merged = new byte[left.length + right.length];
            System.arraycopy(left, 0, merged, 0, left.length);
            System.arraycopy(right, 0, merged, left.length, right.length);
            return merged;
        }
    }

    // 最终压缩器
    static class FinalCompressor {
        public static void compressAll() throws Exception {
            status.set(PipelineStatus.COMPRESSING);
            System.out.println("📦 开始最终压缩...");

            // 模拟最终压缩过程
            Thread.sleep(1000);

            status.set(PipelineStatus.COMPLETED);
            System.out.println("✅ 压缩完成！");
        }
    }

    // 主演示方法
    public static void main(String[] args) throws Exception {
        System.out.println("🚀 启动图片处理管道演示...\n");

        // 初始化状态
        status.set(PipelineStatus.DOWNLOADING);

        // 启动下载线程
        for (int i = 0; i < DOWNLOAD_THREADS; i++) {
            downloadPool.submit(new PicDownloader());
        }

        // 启动处理线程
        for (int i = 0; i < PROCESS_THREADS; i++) {
            processPool.submit(new PicProcessor());
        }

        // 添加下载任务
        System.out.println("📥 添加下载任务...");
        for (int i = 0; i < TOTAL_TASKS; i++) {
            PicTask task = new PicTask("PIC-" + (i + 1), "https://example.com/image" + (i + 1) + ".jpg");
            downloadQueue.put(task);
        }

        // 等待下载完成
        DOWNLOAD_LATCH.await();
        System.out.println("\n📥 所有下载任务完成！");
        status.set(PipelineStatus.PROCESSING);

        // 等待处理完成
        Thread.sleep(2000); // 给处理线程一些时间完成剩余任务

        // 执行最终压缩
        FinalCompressor.compressAll();

        // 关闭线程池
        downloadPool.shutdown();
        processPool.shutdown();
        forkJoinPool.shutdown();

        // 等待所有任务完成
        downloadPool.awaitTermination(10, TimeUnit.SECONDS);
        processPool.awaitTermination(10, TimeUnit.SECONDS);
        forkJoinPool.awaitTermination(10, TimeUnit.SECONDS);

        // 输出统计信息
        printStatistics();

        System.out.println("\n🎉 图片处理管道演示完成！");
    }

    private static void printStatistics() {
        System.out.println("\n📊 ===== 统计信息 =====");
        System.out.printf("总下载大小: %.2f MB%n", totalDownloadSize.get() / 1024.0 / 1024.0);
        System.out.printf("总处理时间: %d ms%n", totalProcessTime.get());
        System.out.printf("缓存命中率: %.1f%%%n", getCacheHitRate() * 100);
        System.out.printf("平均压缩率: %.1f%%%n", getAverageCompressionRate() * 100);
        System.out.printf("最终状态: %s%n", status.get());
    }

    private static double getCacheHitRate() {
        // 简化的缓存命中率计算
        return 0.15; // 假设15%的缓存命中率
    }

    private static double getAverageCompressionRate() {
        // 简化的压缩率计算
        return 0.5; // 假设50%的压缩率
    }
}